package vip.liux.contracts.infrastructure.messaging;

import org.apache.commons.collections4.KeyValue;
import org.apache.commons.collections4.keyvalue.DefaultKeyValue;
import org.springframework.messaging.Message;
import org.springframework.messaging.SubscribableChannel;
import org.springframework.messaging.support.ExecutorSubscribableChannel;
import org.springframework.messaging.support.MessageBuilder;
import vip.liux.contracts.infrastructure.messaging.handlers.MessageHandlerFactory;
import vip.liux.contracts.infrastructure.messaging.interceptors.LoggingChannelInterceptor;
import vip.liux.contracts.infrastructure.messaging.interceptors.RecordChannelInterceptor;
import vip.liux.contracts.shared.enums.Status;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.function.Function;

import static java.util.Optional.ofNullable;

public class MessageBus {

    // 定义一个线程池，用于处理消息的异步发送
    private final static ExecutorService executorService = Executors.newFixedThreadPool(5);

    // 消息总线，存储策略编码与对应的消息通道
    private static final Map<String, KeyValue<MessageTemplateValueObj, SubscribableChannel>> bus = new HashMap<>(128);

    /**
     * 初始化消息总线
     *
     * @param policies 通知策略列表
     */
    public static void init(List<MessageTemplateValueObj> policies) {
        for (MessageTemplateValueObj template : policies) {
            // 如果策略状态为启用，则将其添加到消息总线中
            if (template.getStatus() == Status.ENABLED) {
                bus.put(template.getTemplateCode(), new DefaultKeyValue<>(template, build(template)));
            } else {
                // 如果策略状态为禁用，则从消息总线中移除
                bus.remove(template.getTemplateCode());
            }
        }
    }

    /**
     * 构建消息通道
     *
     * @param dto 通知策略对象
     * @return 构建的消息通道
     */
    protected static ExecutorSubscribableChannel build(MessageTemplateValueObj dto) {
        // 创建一个支持多线程的消息通道
        ExecutorSubscribableChannel channel = new ExecutorSubscribableChannel(executorService);
        // 添加日志拦截器，用于记录消息的发送日志
        channel.addInterceptor(new LoggingChannelInterceptor());
        // 添加记录拦截器，用于记录消息的发送状态
        channel.addInterceptor(new RecordChannelInterceptor());
        // 添加消息处理器，根据通道类型动态构建
        channel.subscribe(MessageHandlerFactory.build(dto.getChannelType()));

        return channel;
    }

    /**
     * 发送消息
     *
     * @param code    策略编码
     * @param message 消息对象
     */
    public static void send(String code, Message<?> message) {
        // 根据策略编码从消息总线中获取通道，并发送消息
        ofNullable(bus.getOrDefault(code, null))
                .ifPresent(c -> c.getValue().send(message));
    }

    /**
     * 发送消息，支持通过函数动态生成消息
     *
     * @param code     策略编码
     * @param supplier 消息生成函数
     */
    public static void send(String code, Function<MessageTemplateValueObj, Message<?>> supplier) {
        // 根据策略编码从消息总线中获取通道，并通过函数生成消息后发送
        ofNullable(bus.getOrDefault(code, null))
                .ifPresent(c -> c.getValue().send(supplier.apply(c.getKey())));
    }

    /**
     * 发送简单文本消息
     *
     * @param code    策略编码
     * @param content 消息内容
     * @param to      消息接收者
     */
    public static void send(String code, String content, String to) {
        // 构建消息对象并发送
        send(code, MessageBuilder.withPayload(content)
                .setHeader(MessageConst.TO, to)
                .build());
    }

    /**
     * 发送带参数的模板消息
     *
     * @param code   策略编码
     * @param params 消息参数
     */
    public static void send(String code, Map<String, Object> params) {
        send(code, dto -> {
            // 获取消息模板内容
            String template = dto.getContent();
            final String[] text = {template};
            // 替换模板中的占位符为实际参数值
            params.forEach((message, value) -> text[0] = text[0].replace("{{" + message + "}}", value.toString()));

            // 构建消息对象
            MessageBuilder<String> builder = MessageBuilder.withPayload(text[0]);
            return builder.build();
        });
    }
}